package com.dmall.scf.action.kanban

import com.dmall.scf.SparkAction
import com.dmall.scf.dto.KanbanFieldValue
import org.apache.spark.sql.types.{LongType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}

import scala.collection.JavaConverters._

/**
 * @descrption
 * scf-spark
 * @author wangxuexing
 * @date 2020/1/10
 */
trait KanbanAction extends SparkAction[KanbanFieldValue] {
  /**
   * 获取datafram
   * @param resultList
   * @param spark
   * @return
   */
  def getDataFrame(resultList: List[KanbanFieldValue], spark: SparkSession): DataFrame= {
    //根据模式字符串生成模式schema
    val fields = List(StructField("company_id", LongType, nullable = false),
      StructField("statistics_date", StringType, nullable = false),
      StructField("field_id", LongType, nullable = false),
      StructField("field_type", StringType, nullable = false),
      StructField("field_value", StringType, nullable = false),
      StructField("other_value", StringType, nullable = false))
    val schema = StructType(fields)
    //将RDD的记录转换为行
    val rowRDD = resultList.map(x=>Row(x.companyId, x.statisticsDate, x.fieldId, x.fieldType, x.fieldValue, x.otherValue)).asJava
    //RDD转为DataFrame
    spark.createDataFrame(rowRDD, schema)
  }

  /**
   * val rowSchema = StructType(Seq(rowSchemaList:_*))
   * //ResultSet反射类对象
   * val rsClass = rs.getClass
   * var count=1;
   * var resultList = new ArrayList[Row]
   * var totalDF = session.createDataFrame(new ArrayList[Row], rowSchema)
   * while (rs.next()) {
   * count=count+1;
   * val temp = new ArrayList[Object]
   * for(i <- 0 to columnTypeList.size()-1){
   * val method = rsClass.getMethod("get"+columnTypeList.get(i),"aa".getClass)
   * temp.add(method.invoke(rs, rsmd.getColumnName(i+1)))
   * }
   * resultList.add(Row(temp:_*))
   * if(count%100000==0){
   * val tempDF = session.createDataFrame(resultList, rowSchema)
   * totalDF=totalDF.union(tempDF).distinct()
   * resultList.clear()
   * }
   * }
   * val tempDF = session.createDataFrame(resultList, rowSchema)
   * totalDF=totalDF.union(tempDF)
   * return totalDF
   * }
   */
  /**
   * 保存mode及表名
   *
   * @return
   */
  override def saveTable: (SaveMode, String) = (SaveMode.Append, "scfc_kanban_field_value")
}
